package com.ziroom.crm.mq.core.processor.retry;

import com.ziroom.crm.mq.core.config.MqConfig;
import com.ziroom.crm.mq.core.context.InvokeContext;
import com.ziroom.crm.mq.core.processor.QueueWrapper;
import com.ziroom.crm.mq.core.processor.SendProcessor;
import com.ziroom.crm.mq.core.retry.*;
import com.ziroom.crm.mq.core.spi.SpiLoader;
import com.ziroom.crm.mq.core.spi.SpiOrder;
import com.ziroom.crm.mq.core.util.AssertUtil;
import lombok.extern.slf4j.Slf4j;

import java.util.Date;
import java.util.Objects;
import java.util.function.Consumer;

/**
 * <p>重试机制</p>
 * 重试的不是具体实现，而是reConsume,reSend 等
 * <p>
 * <PRE>
 * <BR>    修改记录
 * <BR>-----------------------------------------------
 * <BR>    修改日期         修改人          修改内容
 * </PRE>
 *
 * @author WeiShaoying
 * @version 1.0
 * @date Created in 2020年09月08日 10:06
 * @since 1.0
 */
@SpiOrder(SpiOrder.LOWEST_PRECEDENCE - 1)
@Slf4j
public class DefaultSendRetryProcessor implements SendProcessor {

    public static final String GROUP_SEND = "send";

    private final RetryHandler retryHandler;


    public DefaultSendRetryProcessor() {
        // spi 方式加载指定配置
        String retryPolicyName = MqConfig.getConfig(MqConfig.RETRY_SEND_POLICY);
        String delayPolicyName = MqConfig.getConfig(MqConfig.RETRY_SEND_DELAY_POLICY);

        RetryPolicy retryPolicy = SpiLoader.loadPrototypeInstanceByAlias(RetryPolicy.class, retryPolicyName, GROUP_SEND);
        AssertUtil.notNull(retryPolicy, "RetryPolicy with alias[" + retryPolicyName + "] and group [" + GROUP_SEND + "] must has one implement!");

        DelayPolicy delayPolicy = SpiLoader.loadPrototypeInstanceByAlias(DelayPolicy.class, delayPolicyName, GROUP_SEND);
        AssertUtil.notNull(retryPolicy, "DelayPolicy with alias[" + delayPolicyName + "] and group [" + GROUP_SEND + "] must has one implement!");

        retryHandler = new DefaultRetryHandler(retryPolicy, delayPolicy);
    }

    @Override
    public void postSend(InvokeContext invokeContext, QueueWrapper queueWrapper) {
        Consumer<RetryContext> callback = queueWrapper.getRetryCallback();
        if (Objects.isNull(callback)) {
            return;
        }
        if (Objects.isNull(invokeContext.getMessage().getException())) {
            return;
        }
        // 内存重试
        RetryContext retryContext = RetryContext.valueOf(invokeContext, RetryType.IMMEDIATELY, invokeContext.getMessage().getRetryCount());
        // 判断重试次数
        RetryFunction retryFunction = callback::accept;
        retryHandler.retry(retryFunction, retryContext);

        //设置持久化后的重试时间
        if (Objects.nonNull(retryContext.getMessage().getException())) {
            setNextTime(retryContext);
        }
    }

    protected void setNextTime(RetryContext retryContext) {
        Integer secondLater = MqConfig.getConfigInt(MqConfig.RETRY_SEND_SECOND_BEFORE_RESEND, 0);
        retryContext.getMessage().setNextRetryDate(new Date(System.currentTimeMillis() + secondLater * 1000));
    }
}
